#include "message_process_impl.h"
#include "../sync/message_sync_impl.h"
#include <core/system/sleep.h>
#include <core/QbLogger.h>
#include <core/sharedata/login_data.h>
#include <qbprotocol/include/rdMessage.h>
#include <qbmessage/include/message_kernel.h>
#include <string.h>

#ifndef WIN32
#include <arpa/inet.h>
#else
#pragma comment(lib, "ws2_32.lib")
#endif

namespace qb
{
    char CMessageProcessImpl::s_rcvBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN] = {0};
    char CMessageProcessImpl::s_preBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN] = {0};

    CMessageProcessImpl::CMessageProcessImpl()
        : m_event(*EventWrapper::Create(false))
        , m_thread(NULL)
        , m_exit_flag(0)
        , m_spNetworkService(NULL)
        , m_spMessageSync(NULL)
	{
	}

    CMessageProcessImpl::~CMessageProcessImpl()
	{
        delete &m_event;
	}

	STDMETHODIMP_(int32_t) CMessageProcessImpl::Create(THIS_ INetworkService* ins, IMessageSync* sync)
	{
        m_spNetworkService = ins;
        m_spMessageSync = sync;

        m_event.Reset();
        ++m_exit_flag;

        m_thread = ThreadWrapper::CreateThread(ThreadProcess, this, kNormalPriority, "message process thread");
        uint32_t threadid = 0;
        m_thread->Start(threadid);

        return 0;
	}

    STDMETHODIMP_(int32_t) CMessageProcessImpl::Destory(THIS_)
	{ 
		////////////////////////////////////////////////////////////////////////////
		// remark1: 销毁动作必须在主线程中完成, 简化多线程管理逻辑
		// remark2: 这里认为模块的销毁时机一定是在所有调用此模块的子线程结束后
		////////////////////////////////////////////////////////////////////////////

		// 清空观察者
        StopThread(&m_thread);

        m_spMessageSync = NULL;

        return 0;
	}

	STDMETHODIMP_(void) CMessageProcessImpl::UnRegisterMsgProcessorOnThread(int msgType, IMessageProcessorOnSubThread* observer)
	{
        qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
		auto iter = m_threadMsgProcessors.find(msgType);
        if (iter != m_threadMsgProcessors.end())
            m_threadMsgProcessors.erase(iter);
	}

	STDMETHODIMP_(bool) CMessageProcessImpl::RegisterMsgProcessorOnThread(int msgType, IMessageProcessorOnSubThread* observer)
	{
        qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
        auto iter = m_threadMsgProcessors.find(msgType);
        if (iter != m_threadMsgProcessors.end())
        {
            return false;
        }
        m_threadMsgProcessors[msgType] = observer;
        return true;
	}

    STDMETHODIMP_(int32_t) CMessageProcessImpl::RegisterMessageProcessOnMainThreadObserver(THIS_ IMessageProcessOnMainThreadObserver * observer)
    {
        qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
        int32_t nRet = RegisterObserver<IMessageProcessOnMainThreadObserver>(observer,m_observersMainThread);
        if (-1 == nRet)
        {
            log_warning("already enabled");
        }

        return nRet;
    }

    STDMETHODIMP_(int32_t) CMessageProcessImpl::DeRegisterMessageProcessOnMainThreadObserver(THIS_ IMessageProcessOnMainThreadObserver * observer)
    {
        qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
        int32_t nRet = DeRegisterObserver<IMessageProcessOnMainThreadObserver>(observer,m_observersMainThread);
        if (-1 == nRet)
        {
            log_warning("already enabled");
        }

        return nRet;
    }

    STDMETHODIMP_(int32_t) CMessageProcessImpl::PushNetMessage(THIS_ MESSAGE_SHARED_PTR msp)
    {
        m_queue.push(msp);
        m_event.Set();

        return 0;
    }

    STDMETHODIMP_(int32_t) CMessageProcessImpl::PushMainThreadMessage(THIS_ char* buff, int32_t buflen)
    {
        QB_CHECK_RETURN_INT1(buff && buflen > 0);

        xMsgHeadAck_t xHead;
        memcpy(&xHead, buff, sizeof(xHead));
        int msgtype = ntohl(xHead.msgtype);

        // 主线程回调中，默认在回调函数里处理完智能指针的内容，如果需要延长智能指针的生命周期，需要根据协议号，将makeSharedPtr的第三个参数置为true
        //BList<IMessageProcessOnMainThreadObserver*>::iterator iter = m_observersMainThread.begin();
        //for_each_list(m_observersMainThread, iter, (iter++)->first->OnMainThreadProcessData(msgtype, msp));
        return 0;
    }

    void CMessageProcessImpl::StopThread(ThreadWrapper** thread_)
    {
        while (m_exit_flag.Value() > 0)
        {
            --m_exit_flag;
        }
        m_event.Set();

        ThreadWrapper* tmpPtr = *thread_;
        if (tmpPtr)
        {
            tmpPtr->SetNotAlive();

            if (tmpPtr->Stop())
            {
                delete tmpPtr;
            }
        }
    }

    bool CMessageProcessImpl::ThreadProcess(void* args)
    {
        return static_cast<CMessageProcessImpl*>(args)->ThreadProcessMessage(args);
    }

    bool CMessageProcessImpl::ThreadProcessMessage(void* args)
    {
        while (m_exit_flag.Value() > 0)
        {
            if (m_queue.empty())
            {
                m_event.Wait(0xFFFFFFFF);
                if (m_exit_flag.Value() <= 0)
                {
                    break;
                }

                if (m_queue.empty())
                {
                    continue;
                }
            }

            MESSAGE_SHARED_PTR msp = m_queue.pop();
            ProcessMessage(msp->getBuff(), msp->getLen());
            msp.reset();
        }

        return false;
    }

    bool CMessageProcessImpl::ProcessMessage(const char* buff, int buflen)
    {
        QB_CHECK_RETURN_BOOL1(buff && buflen > 0);
        QB_CHECK_RETURN_BOOL1(m_spMessageSync && m_spNetworkService);

        memset(s_rcvBuf, 0x00, Y_RD_SERVICE_MESSAGE_BLOCK_LEN);
        memset(s_preBuf, 0x00, Y_RD_SERVICE_MESSAGE_BLOCK_LEN);
        memcpy(s_rcvBuf, buff, buflen);

        int32_t preLen = 0;
        int retUnPack = rdMessage::iUnPack(s_rcvBuf, buflen, s_preBuf, &preLen);
        if (retUnPack < 0)
        {
            log_warning("Message Unpack Failed!");
            return false;
        }

        xMsgHeadAck_t xHead;
        memcpy(&xHead, s_preBuf, sizeof(xHead));
        xHead.msgtype = ntohl(xHead.msgtype);
        short retCode = ntohs(xHead.retcode);
        {
            qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);

            if (xHead.msgtype == E_FID_CS_FORCE_QUIT)
            {
                m_spNetworkService->NotifyForceQuit(loginData::instance().ConnectID());
                return true;
            }

            auto iter = m_threadMsgProcessors.find(xHead.msgtype);
            if (iter == m_threadMsgProcessors.end() || iter->second == nullptr)
            {
                return NotifyMainThreadObserver(xHead.msgtype, s_preBuf, preLen);
            }
            iter->second->OnResponse(xHead.msgtype, {s_preBuf, preLen, false, retCode});
        }
        return true;
    }

    bool CMessageProcessImpl::NotifyMainThreadObserver(int msgtype, char* buff, int buflen)
    {
        (static_cast<CMessageSyncImpl*>(m_spMessageSync))->PushMessage(buff, buflen);
        return true;
    }
}
